package gd1;

import com.alibaba.fastjson.JSON;
import generator.domain.Orders;
import generator.domain.Payment;
import generator.domain.UserBehaviorLog;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import util.KafkaUtil;

import java.time.Duration;

/**
 * 电商数据流处理主类
 * 功能：将订单数据、支付数据和用户行为数据进行关联分析
 */
public class Dws_Sql {
    public static void main(String[] args) throws Exception {
        // 1. 创建Flink流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 设置并行度为1（生产环境应根据实际情况调整）
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.executeSql("CREATE TABLE dws_orders (\n" +
                "  `order_id` BIGINT,\n" +
                "  `user_id` BIGINT,\n" +
                "  `product_id` STRING,\n" +
                "  product_name string,\n" +
                "  order_status string,\n" +
                "  quantity int,\n" +
                "  total_amount double,\n" +
                "  payment_amount double,\n" +
                "  buyer_count int,\n" +
                "  device_type string,\n" +
                "  create_time bigint,\n" +
                "  time_ltz AS TO_TIMESTAMP_LTZ(create_time, 3),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '0' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'dwd_order_09_06',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")");
        tenv.sqlQuery("select * from dws_orders").execute().print();
        tenv.executeSql("CREATE TABLE dws_payment (\n" +
                "  `payment_id` BIGINT,\n" +
                "  `order_id` BIGINT,\n" +
                "  `user_id` bigint,\n" +
                "  payment_amount double,\n" +
                "  payment_status string,\n" +
                "  buyer_count int,\n" +
                "  device_type string,\n" +
                "  create_time bigint,\n" +
                "  time_ltz AS TO_TIMESTAMP_LTZ(create_time, 3),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '0' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'dwd_payment_09_06',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")");
        tenv.executeSql("CREATE TABLE dws_user_log (\n" +
                "  `log_id` BIGINT,\n" +
                "  `user_id` BIGINT,\n" +
                "  `behavior_type` STRING,\n" +
                "  `page_views` int,\n" +
                "  visitor_count int,\n" +
                "  cart_count int,\n" +
                "  device_type string,\n" +
                "  referrer string,\n" +
                "  `create_time` bigint,\n" +
                "  time_ltz AS TO_TIMESTAMP_LTZ(create_time, 3),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '0' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'dwd_user_09_06',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")");
//        tenv.sqlQuery("select * from dws_user_log").execute().print();
        tenv.executeSql("CREATE TABLE ads_user_log (\n" +
                "  `log_id` BIGINT,\n" +
                "  `user_id` BIGINT,\n" +
                "  `behavior_type` STRING,\n" +
                "  `page_views` int,\n" +
                "  visitor_count int,\n" +
                "  cart_count int,\n" +
                "  device_type string,\n" +
                "  referrer string,\n" +
                "  `create_time` bigint,\n" +
                "  time_ltz AS TO_TIMESTAMP_LTZ(create_time, 3),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '0' SECOND\n" +
                ")WITH (\n" +
                "  'connector'='doris',\n" +
                "    'fenodes'='node102:8030',\n" +
                "    'table.identifier'='gd3.ads_user_log',\n" +
                "    'username'='root',\n" +
                "    'password'='123456',\n" +
                "    'sink.label-prefix'='doris_label'\n" +
                ")");


        tenv.executeSql("CREATE TABLE ads_payment (\n" +
                "  `payment_id` BIGINT,\n" +
                "  `order_id` BIGINT,\n" +
                "  `user_id` bigint,\n" +
                "  payment_amount double,\n" +
                "  payment_status string,\n" +
                "  buyer_count int,\n" +
                "  device_type string,\n" +
                "  create_time bigint,\n" +
                "  time_ltz AS TO_TIMESTAMP_LTZ(create_time, 3),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '0' SECOND\n" +
                ") WITH (\n" +
                "  'connector'='doris',\n" +
                "    'fenodes'='node102:8030',\n" +
                "    'table.identifier'='gd3.ads_payment',\n" +
                "    'username'='root',\n" +
                "    'password'='123456',\n" +
                "    'sink.label-prefix'='doris_label'\n" +
                ")");

        tenv.executeSql("CREATE TABLE ads_orders (\n" +
                "  `order_id` BIGINT,\n" +
                "  `user_id` BIGINT,\n" +
                "  `product_id` STRING,\n" +
                "  product_name string,\n" +
                "  order_status string,\n" +
                "  quantity int,\n" +
                "  total_amount double,\n" +
                "  payment_amount double,\n" +
                "  buyer_count int,\n" +
                "  device_type string,\n" +
                "  create_time bigint,\n" +
                "  time_ltz AS TO_TIMESTAMP_LTZ(create_time, 3),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '0' SECOND\n" +
                ") WITH (\n" +
                "  'connector'='doris',\n" +
                "    'fenodes'='node102:8030',\n" +
                "    'table.identifier'='gd3.ads_orders',\n" +
                "    'username'='root',\n" +
                "    'password'='123456',\n" +
                "    'sink.label-prefix'='doris_label'\n" +
                ")");
//        tenv.createTemporaryView("dws_orders",tableResult);
//        tenv.executeSql("insert into ads_orders" +
//                "select order_id," +
//                "user_id," +
//                "product_id," +
//                "product_name," +
//                "order_status," +
//                "quantity," +
//                "total_amount," +
//                "payment_amount," +
//                "buyer_count," +
//                "device_type," +
//                "create_time," +
//                "time_ltz" +
//                " from dws_orders");
//        tenv.createTemporaryView("dws_payment",tableResult);
//        tenv.executeSql("insert into ads_payment" +
//                "select payment_id," +
//                "order_id," +
//                "user_id," +
//                "payment_amount," +
//                "payment_status," +
//                "buyer_count," +
//                "device_type," +
//                "create_time," +
//                "time_ltz from dws_payment");
//        tenv.createTemporaryView("dws_user_log",tableResult);
//        tenv.executeSql("insert into ads_user_log" +
//                "select log_id," +
//                "user_id," +
//                "behavior_type," +
//                "page_views," +
//                "visitor_count," +
//                "cart_count," +
//                "device_type," +
//                "referrer," +
//                "create_time," +
//                "time_ltz from dws_user_log");



/*

 */














    }
}